package com.itheima.kafka.consumer.handler;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;

/**
 * Kafka消费者处理类
 */
@Component
public class KafkaConsumerHandler {
    
    // 定义日志记录器
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerHandler.class);
    
    // @KafkaListener(topics = "itheima", groupId = "itheima.demo")
    public void consumerHandler(String msg) {
        LOGGER.info("consumer topic is : {} , msg is ----> {} ", "itheima", msg);
    }
    
    // @KafkaListener(topics = "itheima", groupId = "itheima.demo")
    public void consumerNotRepeatHandler(String msg, KafkaConsumer consumer) {
        LOGGER.info("consumer topic is {},msg is ---->{}", "itheima", msg);
        
        OffsetCommitCallback callback = new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                if (e == null) {
                    LOGGER.info("位移提交成功");
                    return;
                }
                Set<Map.Entry<TopicPartition, OffsetAndMetadata>> entries = map.entrySet();
                for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : entries) {
                    OffsetAndMetadata offsetAndMetadata = entry.getValue();
                    TopicPartition topicPartition = entry.getKey();
                    LOGGER.info("位移提交失败,topicPartition is {},offsetAndMetadata is {}",
                            topicPartition, offsetAndMetadata);
                }
            }
        };
        //异步位移提交
        consumer.commitAsync(callback);
    }
    
    public void consumerOrderMessageHandler1(String msg, KafkaConsumer
            consumer) {
        LOGGER.info("consumer topic is : {} , msg is ----> {} ", "itcast",
                msg);
        consumer.commitAsync();
    }
    
    @KafkaListener(topicPartitions =
            {@org.springframework.kafka.annotation.TopicPartition(topic = "itcast",
                    partitions = "0")}, groupId = "itcast.demo")
    public void consumerOrderMessageHandler(String msg, KafkaConsumer consumer) {
        LOGGER.info("consumer topic is {},msg is ---->{}", "itcast", msg);
        
        OffsetCommitCallback callback = (map, e) -> {
            if (e == null) {
                LOGGER.info("提交成功");
                return;
            }
            Set<Map.Entry<TopicPartition, OffsetAndMetadata>> entries = map.entrySet();
            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : entries) {
                OffsetAndMetadata offsetAndMetadata = entry.getValue();
                TopicPartition topicPartition = entry.getKey();
                LOGGER.info("提交失败,topicPartition is {},offsetAndMetadata is {}",
                        topicPartition, offsetAndMetadata);
            }
        };
        //异步位移提交
        consumer.commitAsync(callback);
    }
    
    /**
     * 指定分区消费
     */
    // @KafkaListener(topicPartitions = {
    //         @org.springframework.kafka.annotation.TopicPartition(topic = "itcast",
    //                 partitions = "0")
    // }, groupId = "itcast.demo")
    public void consumerOrderMessageHandler2(String msg, KafkaConsumer consumer) {
        LOGGER.info(" consumer topic is :{},msg is --->{}", "itcast", msg);
        //异步位移提交
        consumer.commitAsync();
    }
}
